package com.ehotting.mqbd.core.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.ehotting.eaf.core.DateHelper;
import com.ehotting.eaf.core.utils.SpringUtils;
import com.ehotting.eaf.netty.domain.Request;
import com.ehotting.mqbd.api.MqDataStateConstants;
import com.ehotting.mqbd.api.TaskStateActionConstants;
import com.ehotting.mqbd.api.bean.MQBaseData;
import com.ehotting.mqbd.api.bean.MQDataAble;
import com.ehotting.mqbd.api.bean.MsgStateBean;
import com.ehotting.mqbd.api.enums.MsgStateEnum;
import com.ehotting.mqbd.api.enums.ProtocolTypeEnum;
import com.ehotting.mqbd.api.enums.TopicEnum;
import com.ehotting.mqbd.core.helper.MQStringHelper;
import com.ehotting.mqbd.core.manager.AgentLinkManager;
import com.ehotting.mqbd.core.manager.MsgStateManager;
import com.ehotting.mqbd.core.mq.consumer.ClientMQDataHandlerAble;
import com.ehotting.mqbd.core.service.TaskStateMQService;
import com.ehotting.mqbd.core.service.mq.MQDataHelper;
import com.ehotting.mqbd.core.sync.MqSyncFutureMapHelper;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Date;
import java.util.List;


public class MessageProducer {

    private Logger logger = LoggerFactory.getLogger(MessageProducer.class);

    private DefaultMQProducer producer;
    private String producerGroup;
    private MessageProducerParam messageProducerParam;
    @Autowired
    private MsgStateManager msgStateManager;
    @Autowired
    private MqSyncFutureMapHelper<MQDataAble> mqSyncFutureMapHelper;
    @Autowired
    private AgentLinkManager agentLinkManager;
    @Autowired
    private TaskStateMQService taskStateMQService;
    @Autowired
    protected MQDataHelper mqDataHelper;


    public MessageProducer(MessageProducerParam messageProducerParam,String producerGroup) {
        this.messageProducerParam = messageProducerParam;
        this.producerGroup = producerGroup;
    }

    public void start() throws MQClientException {

        producer = new DefaultMQProducer(producerGroup);
        logger.info("messageProducerParam:{}",JSON.toJSONString(messageProducerParam));
        // 设置 NameServer 地址
        producer.setNamesrvAddr(messageProducerParam.getNameSrvAddr());
        producer.setRetryTimesWhenSendFailed(messageProducerParam.getRetryTimesWhenSendFailed());
        producer.setSendMsgTimeout(messageProducerParam.getSendMsgTimeout());
        producer.setMaxMessageSize(messageProducerParam.getMaxMessageSize());
        // 启动 producer
        producer.start();
    }

    public SendResult send(MQDataAble data)  {

        if(isSolvingMqData(data)){
            logger.error("mqData[{}] is solving.return.",data.getAgentId());
            return null;
        }


        if(!mqDataHelper.isSend(data)){
            return null;
        }

        SendResult sendResult = null;

        try {
            String topicName = MQStringHelper.createProducerTopic(data.getTopic(),data) ;
            logger.info("topicName:{}",topicName);
            String tag = MQStringHelper.createProducerTag(data) ;
            logger.info("tag:{}",tag);

            if(data.isSendRequest()==false){
                ((MQBaseData<Request, Request>)data).setRequest(null);
            }

            if(data.isSendResponse()==false){
                ((MQBaseData<Request, Request>)data).setResponse(null);
            }

            String msg = JSON.toJSONString(data, SerializerFeature.DisableCircularReferenceDetect);
            logger.info("msg:{}",msg);
            Message message = new Message(
                    topicName,
                    tag,
                    data.getSerial(),
                    msg.getBytes(RemotingHelper.DEFAULT_CHARSET));


            if(data.getTopic() == TopicEnum.BLOCK){
                sendResult = producer.send(message,new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        int index = 0;
                        return mqs.get(index);
                    }
                }, 0);
                if(SendStatus.SEND_OK == sendResult.getSendStatus()){
                    data.setMsgId(sendResult.getMsgId());
                    msgStateManager.put(sendResult.getMsgId(),data.getAgentId(),data.getSerial());
                }
            }else{
                sendResult = producer.send(message);
            }

            logger.info("sendResult:{}",JSON.toJSONString(sendResult));
            setReturnMsg( sendResult, data);
        } catch (Exception e) {
            data.setState(MqDataStateConstants.SEND_EXCEPTION);
            data.setMsg("消息发送异常");
            logger.error("MessageProducer send exception",e);
        }
        return sendResult;
    }

    private void setReturnMsg(SendResult sendResult,MQDataAble data){
        String time = "["+DateHelper.toSeparatorStringyyyyMMddHHmmss(new Date())+"]";
        if(SendStatus.SEND_OK == sendResult.getSendStatus()){
            data.setState(MqDataStateConstants.SEND_SUCCESS);
            data.setMsg(time+"消息发送成功");
            if(data.getTopic()== TopicEnum.BLOCK) {
                long index = msgStateManager.getQueueIndexByMsgId(data.getModule(), data.getMsgId());

                String moduleName = "任务";

                Long mainProjectId = agentLinkManager.getProjectId(data.getAgentId());

                String projectNo = "";
                if(mainProjectId>0){
                    projectNo = "[编号:"+mainProjectId+"]";
                }
                String msg = time + moduleName + projectNo+"排队为第 "+index+" 位";

                if(index<=0){
                    msg = time + moduleName + projectNo+"排队为第 1 位";
                }

                ClientMQDataHandlerAble clientMQDataHandler = SpringUtils.getBean(ClientMQDataHandlerAble.class);
                if(clientMQDataHandler!=null){
                    clientMQDataHandler.invoke(data.getAgentId(),
                            taskStateMQService.createTaskState(
                                    TaskStateActionConstants.TASK_QUEUE,
                                    msg,
                                    data.getAgentId()));
                }

            }
        }else{
            data.setState(MqDataStateConstants.SEND_FAIL);
            data.setMsg(time+"消息发送失败");
        }
    }


    public void shutdown(){
        producer.shutdown(); //（5）
    }


    public boolean isSolvingMqData(MQDataAble data){
        if(data.getTopic() == TopicEnum.CONCURRENCY ) {
            return false;
        }

        /**HTTP求解可以重复提交**/
        if(data.getProtocolType() == ProtocolTypeEnum.HTTP ) {
            return false;
        }

        MsgStateBean msgState =  msgStateManager.getByAgentIdAndInLive(data.getAgentId());
        if(msgState == null){
            return false;
        }

        String time = "["+DateHelper.toSeparatorStringyyyyMMddHHmmss(new Date())+"]";
        if(msgState.getState() == MsgStateEnum.QUEUE){
            long queueIndex = msgStateManager.getQueueIndexByAgentId(data.getModule(), data.getAgentId());
            logger.info("queueIndex:{}",queueIndex);
            if(queueIndex>0){
                data.setState(MqDataStateConstants.EXISTS_IN_QUEUE);
                data.setMsg(time+"任务已经在排队第 "+ queueIndex +" 位，请不要重复提交。");
            }else{
                //data.setMsg("任务正准备求解，请不要重复提交。");
                logger.error("任务状态还在队列中，但是已经出队");
                return false;
            }
        }else if(msgState.getState()== MsgStateEnum.SOLVING){
            //data.setState(MqDataStateConstants.EXISTS_IN_SOLVING);
            //data.setMsg("任务已经在求解，请不要重复提交。");
            logger.error(time+"任务已经在求解，可能正在取消，请等待再提交。");
            return false;
        }

        mqSyncFutureMapHelper.putAndNotify(data.getSerial(),null);

        return true;
    }

    public static class MessageProducerParam{

        private String nameSrvAddr;
        private Integer maxMessageSize;
        private Integer sendMsgTimeout;
        private Integer retryTimesWhenSendFailed;

        public String getNameSrvAddr() {
            return nameSrvAddr;
        }

        public void setNameSrvAddr(String nameSrvAddr) {
            this.nameSrvAddr = nameSrvAddr;
        }

        public Integer getMaxMessageSize() {
            return maxMessageSize;
        }

        public void setMaxMessageSize(Integer maxMessageSize) {
            this.maxMessageSize = maxMessageSize;
        }

        public Integer getSendMsgTimeout() {
            return sendMsgTimeout;
        }

        public void setSendMsgTimeout(Integer sendMsgTimeout) {
            this.sendMsgTimeout = sendMsgTimeout;
        }

        public Integer getRetryTimesWhenSendFailed() {
            return retryTimesWhenSendFailed;
        }

        public void setRetryTimesWhenSendFailed(Integer retryTimesWhenSendFailed) {
            this.retryTimesWhenSendFailed = retryTimesWhenSendFailed;
        }
    }
}
